package com.uuuo.rxjava1;

import org.junit.Test;

import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.schedulers.Schedulers;

/**
 * Created by JiaTao on 2017/4/5.
 */

public class D_Combine extends A_Base {
    Observable<Integer> ob1 = Observable.create(new Observable.OnSubscribe<Integer>() {
        @Override
        public void call(Subscriber<? super Integer> subscriber) {
            for (int i = 0; i < 10; i++) {
                try {
                    subscriber.onNext(i);
                    Thread.sleep(300);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    });
    Observable<Integer> ob2 = Observable.create(new Observable.OnSubscribe<Integer>() {
        @Override
        public void call(Subscriber<? super Integer> subscriber) {
            for (int i = 500; i < 510; i++) {
                try {
                    subscriber.onNext(i);
                    Thread.sleep(150);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    });


    @Test
    public void startWith() {
        Observable.just(1, 2, 3, 4, 5)
                .startWith(8, 9)
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        showMsg(integer + "");
                    }
                });
    }

    @Test
    public void merge() {
        Observable.merge(ob2, ob1).subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer integer) {
                showMsg(integer + "");
            }
        });

    }

    @Test
    public void zip() {
        Observable.zip(ob1, ob2, new Func2<Integer, Integer, String>() {
            @Override
            public String call(Integer integer, Integer integer2) {
                return integer + "----" + integer2;
            }
        }).subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                showMsg(s);
            }
        });
    }

    @Test
    public void join() {
        Observable create1 = Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                for (int i = 0; i < 6; i++) {
                    subscriber.onNext(i);

                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        subscriber.onError(e);
                    }
                }
            }
        }).subscribeOn(Schedulers.newThread());

        Observable create2 = Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                for (int i = 0; i < 4; i++) {
                    subscriber.onNext(i);

                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        subscriber.onError(e);
                    }
                }
            }
        }).subscribeOn(Schedulers.newThread());

        create1.join(create2,
                new Func1<Integer, Observable<Long>>() {
                    @Override
                    public Observable<Long> call(Integer integer) {
                        return Observable.timer(1000, TimeUnit.MILLISECONDS);
                    }
                },
                new Func1<Integer, Observable<Long>>() {
                    @Override
                    public Observable<Long> call(Integer integer) {
                        return Observable.timer(1000, TimeUnit.MILLISECONDS);
                    }
                },
                new Func2<Integer, Integer, String>() {
                    @Override
                    public String call(Integer integer1, Integer integer2) {
                        return integer1 + "-" + integer2;
                    }
                })
                .subscribe(new Subscriber<String>() {

                    @Override
                    public void onCompleted() {
                        System.out.println("onCompleted.");
                    }

                    @Override
                    public void onError(Throwable e) {
                        System.out.println("onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(String s) {
                        System.out.println("onNext: " + s);
                    }
                });
    }
}
